package action.chap7;

import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.ForkJoinTask;
import java.util.function.Function;
import java.util.stream.LongStream;
import java.util.stream.Stream;

public class ParallelStreams {
    public static long measureSumPerf(Function<Long, Long> adder, long n) {
        long fastest = Long.MAX_VALUE;
        for (int i = 0; i < 10; i++) {
            long start = System.nanoTime();
            long sum = adder.apply(n);
            long duration = (System.nanoTime() - start) / 1_000_000;
            System.out.println("Result: " + sum);
            if (duration < fastest) fastest = duration;
        }
        return fastest;
    }

    /**
     * 实验方法      最短执行时间   主线程执行时间
     * sequentialSum  90msecs   1412     iterate生成的是装箱的对象，必须拆箱成数字才能求和；
     * iterativeSum   3 msecs   105      用传统for循环的迭代版本执行起来应该会快很多，因为它更为底层，更重要的是不需要对原始类型做任何装箱或拆箱操作
     * parallelSum    128 msecs 5795     存在拆箱装箱操作
     * rangedSum      4 msecs   145  LongStream.rangeClosed直接产生原始类型的long数字，没有装箱拆箱的开销。  数字流
     * parallelRangedSum  1 msecs 13    并行归纳，比较快的操作  没有拆箱装箱操作的并行版本
     * sideEffectSum    4 msecs   151    单线程速度相对还可以
     *  此次实现 是计算不消耗太多时间资源的基础上做的
     * 总结实验数据表明
     *  1、把顺序流转成并行流轻而易举，但却不一定是好事。已经指出，并行流并不总是比顺序流快。
     *  2、留意装箱。自动装箱和拆箱操作会大大降低性能
     *      Java 8中有原始类型流（IntStream、LongStream、DoubleStream）来避免这种操作，但凡有可能都应该用这些流。
     *  3、有些操作本身在并行流上的性能就比顺序流差。特别是limit和findFirst等依赖于元
     *      素顺序的操作，它们在并行流上执行的代价非常大。例如，findAn 会比findFirst
     *      能好，因为它不一定要按顺序来执行。你总是可以调用unordered方法来把有序流变成无序流。
     *  4、要考虑流背后的数据结构是否易于分解。 如，ArrayList的拆分效 比LinkedList高得多，因为前者用不着遍历就可以平均拆分，而后者则必须遍历。
     *
     * @param args
     */
    public static void main(String[] args) {
        long start = System.currentTimeMillis();
        System.out.println("Sequential sum done in:" +
                measureSumPerf(ParallelStreams::forkJoinSum, 10_000_000) + " msecs");
        //测试递归
        //sequentialSum(10_000_000);
        //iterativeSum(10_000_000);
        //并行版本
       // long l = parallelSum(10_000_000);
       // System.out.println(l);
        long end =  System.currentTimeMillis();
        System.out.println(end-start);
    }

    /**
     *  用流便利性能是比较低的  迭代器是需要拆箱装箱的
     * @param n
     * @return
     */
    public static long sequentialSum(long n) {
        //iterate生成的是装箱的对象，必须拆箱成数字才能求和；
        return Stream.iterate(1L, i -> i + 1)
                .limit(n)
                .reduce(0L, Long::sum);
    }



    /**
     * 用传统for循环的迭代版本执行起来应该会快很多，因为它更为底层，更重要的是不需要对原始类型做任何装箱或拆箱操作
     * @param n
     * @return
     */
    public static long iterativeSum(long n) {
        long result = 0;
        for (long i = 1L; i <= n; i++) {
            result += i;
        }
        return result;
    }

    /**
     *  并行版本
     * @param n
     * @return
     */
    public static long parallelSum(long n) {
        //我们很难把iterate分成多个独立块来并行执行。
        return Stream.iterate(1L, i -> i + 1)
                .limit(n)
                .parallel()
                .reduce(0L, Long::sum);
    }
    //使用数字流进行操作的
    public static long rangedSum(long n) {

        return LongStream.rangeClosed(1, n)
                .reduce(0L, Long::sum);
    }


    /**
     * 并行归纳，比较快的操作
     * 这是一个比顺序归纳更快的一个操作哟
     * @param n
     * @return
     */
    public static long parallelRangedSum(long n) {
        return LongStream.rangeClosed(1, n)
                .parallel()
                .reduce(0L, Long::sum);
    }
    public static long sideEffectSum(long n) {
        Accumulator accumulator = new Accumulator();
        LongStream.rangeClosed(1, n).forEach(accumulator::add);
        return accumulator.total;
    }

    /**
     * accumulator::add里面使用的算法存在线程安全的问题使用并行流不能保证正确性
     * @param n
     * @return
     */
    public static long sideEffectParallelSum(long n) {
        Accumulator accumulator = new Accumulator();
        LongStream.rangeClosed(1, n).parallel().forEach(accumulator::add);

        return accumulator.total;
    }

    public static long forkJoinSum(long n) {
        long[] numbers = LongStream.rangeClosed(1, n).toArray();
        ForkJoinTask<Long> task = new ForkJoinSumCalculator(numbers,0,numbers.length);
        //  使用ForkJoinPool().invoke(task) 无论在什么时候都不能在任务内部调用 invoke
        return new ForkJoinPool().invoke(task);
    }
}


 class Accumulator {
    public Long total = 0L;
     public void add(long value) {
         total+=value;
     }


}
